package com.fx.smsclient.util;

import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.ObjectUtil;
import com.zx.sms.BaseMessage;
import com.zx.sms.common.util.ChannelUtil;
import com.zx.sms.connect.manager.CMPPEndpointManager;
import com.zx.sms.connect.manager.EndpointEntity;
import io.netty.channel.ChannelFuture;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import io.netty.util.concurrent.Promise;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

@Component
public class SendMsgUtil {
    private static final Logger logger = LoggerFactory.getLogger(SendMsgUtil.class);

    static private HashMap<String,LinkedBlockingQueue<BaseMessage>> endPointMsgQueue = new HashMap<>();

    static private ExecutorService executor = ThreadUtil.newExecutor();;

    static public void putMsg(String endpointId, BaseMessage msg){
        try {
            if (endPointMsgQueue.containsKey(endpointId)){
                LinkedBlockingQueue<BaseMessage> msgQueue =endPointMsgQueue.get(endpointId);
                msgQueue.put(msg);
            }else {
                logger.info("endpointId:{} add",endpointId);
                LinkedBlockingQueue<BaseMessage> msgQueue = new LinkedBlockingQueue<>();
                endPointMsgQueue.put(endpointId,msgQueue);
                msgQueue.put(msg);
                runTreads(endpointId);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }

    static private void runTreads(String endpointId) {
        executor.execute(() -> {
            LinkedBlockingQueue<BaseMessage> msgQueue= endPointMsgQueue.get(endpointId);
            logger.info("线程启动");
            while (true) {
                try {
                    BaseMessage msg = msgQueue.take();
                    sendMsg(endpointId, msg);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
    }

    static private void sendMsg(String endpointId, BaseMessage msg){
        boolean isOk =false;
        while (!isOk){
            isOk= sendMsgToChannel(endpointId, msg);
            if (isOk) break;
            try {
                Thread.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public static List<BaseMessage> sendMsg2(String endpointId, BaseMessage msg){
        boolean isOk =false;
        while (!isOk){
            List<BaseMessage> ret= sendSyncMsgToChannel(endpointId, msg);
            if (ObjectUtil.isNotEmpty(ret)) {
                return ret;
            }
            try {
                Thread.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        return null;
    }

    static private boolean sendMsgToChannel(String endpointId, BaseMessage msg)  {
        logger.info("endpointId:{}",endpointId);
        EndpointEntity endpointEntity = CMPPEndpointManager.INS.getEndpointEntity(endpointId);
        if (ObjectUtil.isEmpty(endpointEntity)) {
            logger.error("endpointId:{}不存在",endpointId);
            return true;
        }
        ChannelFuture future = ChannelUtil.asyncWriteToEntity(endpointEntity, msg);
        if (future != null) {
            try {
                future.addListener(new GenericFutureListener<Future<? super Void>>() {
                    @Override
                    public void operationComplete(Future<? super Void> future) throws Exception {
                    }
                });
            } catch (Exception e) {
                e.printStackTrace();
            }
            return true;
        }else {
            logger.error("futures is null!!");
            return false;
        }
    }

    static private List<BaseMessage> sendSyncMsgToChannel(String endpointId, BaseMessage msg)  {
        logger.info("endpointId:{}",endpointId);
        EndpointEntity endpointEntity = CMPPEndpointManager.INS.getEndpointEntity(endpointId);
        if (ObjectUtil.isEmpty(endpointEntity)) {
            logger.error("endpointId:{}不存在",endpointId);
            return null;
        }
        Promise<BaseMessage> frefuture = null;
        List<Promise<BaseMessage>> futures = null;
        ChannelFuture chfuture = null;
        List<BaseMessage> retList = new ArrayList<>() ;
        try {
            futures = ChannelUtil.syncWriteLongMsgToEntity(endpointId, msg);
            if (chfuture != null)
                chfuture.sync();
            if (futures != null) {
                for ( Promise<BaseMessage> future:futures){
                    retList.add(future.get(5000, TimeUnit.MILLISECONDS));
                }
                return retList;
            }else {
                logger.error("futures is null!!");
                return null;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

}
